package com.ihr360.job.core.item;

import com.ihr360.job.core.entity.StepContribution;
import com.ihr360.job.core.job.builder.JobBuilderHelper;
import com.ihr360.job.core.repeat.RepeatStatus;
import com.ihr360.job.core.scope.context.ChunkContext;
import com.ihr360.job.core.scope.context.SimpleStepContext;
import com.ihr360.job.core.scope.context.StepContext;
import com.ihr360.job.core.step.tasklet.Tasklet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A {@link Tasklet} implementing variations on read-process-write item
 * handling.
 *
 * @param <I> input item type
 * @author Dave Syer
 */
public class ChunkOrientedTasklet<I> implements Tasklet {
    protected final Logger logger = LoggerFactory.getLogger(JobBuilderHelper.class.getName());
    private static final String INPUTS_KEY = "INPUTS";

    private final ChunkProcessor<I> chunkProcessor;

    private final ChunkProvider<I> chunkProvider;

    private boolean buffering = true;

    public ChunkOrientedTasklet(ChunkProvider<I> chunkProvider, ChunkProcessor<I> chunkProcessor) {
        this.chunkProvider = chunkProvider;
        this.chunkProcessor = chunkProcessor;
    }

    /**
     * Flag to indicate that items should be buffered once read. Defaults to
     * true, which is appropriate for forward-only, non-transactional item
     * readers. Main (or only) use case for setting this flag to false is a
     * transactional JMS item reader.
     *
     * @param buffering
     */
    public void setBuffering(boolean buffering) {
        this.buffering = buffering;
    }

    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {

        @SuppressWarnings("unchecked")
        Chunk<I> inputs = (Chunk<I>) chunkContext.getAttribute(INPUTS_KEY);

        StepContext stepContext = chunkContext.getStepContext();
        Long stepExecutionId = stepContext.getStepExecution().getId();
        Long jobExecutionId = stepContext.getStepExecution().getJobExecutionId();
        SimpleStepContext simpleStepContext = new SimpleStepContext(jobExecutionId,stepExecutionId);

        if (inputs == null) {
            inputs = chunkProvider.provide(contribution,simpleStepContext);
            if (buffering) {
                chunkContext.setAttribute(INPUTS_KEY, inputs);
            }
        }

        chunkProcessor.process(contribution, inputs,simpleStepContext);
        chunkProvider.postProcess(contribution, inputs);

        // Allow a message coming back from the processor to say that we
        // are not done yet
        if (inputs.isBusy()) {
            logger.debug("Inputs still busy");
            return RepeatStatus.CONTINUABLE;
        }

        chunkContext.removeAttribute(INPUTS_KEY);
        chunkContext.setComplete();

        if (logger.isDebugEnabled()) {
            logger.debug("Inputs not busy, ended: " + inputs.isEnd());
        }
        return RepeatStatus.continueIf(!inputs.isEnd());

    }

}
